package queue

import (
	"net/url"
	"sync"

	"gitee.com/i2worker/colite"
)

// Storage 队列存储器接口
type Storage interface {
	// Init 初始化
	Init() error

	// AddRequest 向队列中添加序列化的Request
	AddRequest([]byte) error

	// GetRequest 从队列中获取序列化的Request
	GetRequest() ([]byte, error)

	// QueueSize 队列大小
	QueueSize() (int, error)
}

// 请求队列(并发模式)
type Queue struct {
	// 并发请求数
	Threads int

	storage Storage
	wake    chan struct{}
	mut     sync.Mutex
}

// New 创建请求队列
func New(threads int, s Storage) (*Queue, error) {
	if s == nil {
		s = &InMemoryQueueStorage{MaxSize: 100000}
	}
	if err := s.Init(); err != nil {
		return nil, err
	}
	return &Queue{
		Threads: threads,
		storage: s,
	}, nil
}

// AddURL 添加一个网址到队列中
func (q *Queue) AddURL(URL string) error {
	u, err := url.Parse(URL)
	if err != nil {
		return err
	}
	r := &colite.Request{
		URL:    u,
		Method: "GET",
	}
	d, err := r.Marshal()
	if err != nil {
		return err
	}
	return q.storage.AddRequest(d)
}

// AddRequest 添加一个Request到队列中
func (q *Queue) AddRequest(r *colite.Request) error {
	q.mut.Lock()
	waken := q.wake != nil
	q.mut.Unlock()
	if !waken { //队列还没有启动
		return q.storeRequest(r)
	}
	err := q.storeRequest(r)
	if err != nil {
		return err
	}
	q.wake <- struct{}{} //向通道写入添加请求信息
	return nil
}

// storeRequest 保存Request
func (q *Queue) storeRequest(r *colite.Request) error {
	d, err := r.Marshal()
	if err != nil {
		return err
	}
	return q.storage.AddRequest(d)
}

// Run 启动线程并调用Collector抓取数据
func (q *Queue) Run(c *colite.Collector) error {
	// 检测队列是否已经启动
	q.mut.Lock()
	if q.wake != nil {
		q.mut.Unlock()
		panic("Cannot call duplicate Queue.Run")
	}
	q.wake = make(chan struct{})
	q.mut.Unlock()

	requestc := make(chan *colite.Request) //请求通道
	complete := make(chan struct{})        //完成通道
	errc := make(chan error, 1)            //错误通道

	// 按照并发数启动队列
	for i := 0; i < q.Threads; i++ {
		go independentRunner(requestc, complete)
	}

	// 轮询通道直到抓取完成
	go q.loop(c, requestc, complete, errc)
	defer close(requestc)
	return <-errc
}

// independentRunner 执行请求
func independentRunner(requestc <-chan *colite.Request, complete chan<- struct{}) {
	for req := range requestc {
		req.Do()
		complete <- struct{}{} //向通道写入完成信息
	}
}

// loop 轮询通道
func (q *Queue) loop(c *colite.Collector, requestc chan<- *colite.Request, complete <-chan struct{}, errc chan<- error) {
	var active int
	for {
		size, err := q.storage.QueueSize() //获取队列大小
		if err != nil {
			errc <- err
			break
		}
		if size == 0 && active == 0 { //队列为空且无请求时终止
			errc <- nil
			break
		}
		sent := requestc
		var req *colite.Request
		if size > 0 {
			req, err = q.loadRequest(c) //从队列中取出一个请求
			if err != nil {
				continue
			}
		} else {
			sent = nil
		}
	Sent:
		for {
			select {
			case sent <- req: //执行请求
				active++
				break Sent
			case <-q.wake: //添加新请求
				if sent == nil {
					break Sent
				}
			case <-complete: //完成
				active--
				if sent == nil && active == 0 {
					break Sent
				}
			}
		}
	}
}

// loadRequest 从队列中载入一个请求
func (q *Queue) loadRequest(c *colite.Collector) (*colite.Request, error) {
	buf, err := q.storage.GetRequest()
	if err != nil {
		return nil, err
	}
	copied := make([]byte, len(buf))
	copy(copied, buf)
	return c.UnmarshalRequest(copied)
}

// IsEmpty 判断队列是否为空
func (q *Queue) IsEmpty() bool {
	s, _ := q.Size()
	return s == 0
}

// Size 获取队列的大小
func (q *Queue) Size() (int, error) {
	return q.storage.QueueSize()
}

// InMemoryQueueStorage 保存在内存中的请求队列
type InMemoryQueueStorage struct {
	// MaxSize 请求容量
	MaxSize int

	first *inMemoryQueueItem //第一个请求
	last  *inMemoryQueueItem //最后一个请求
	size  int                //队列大小
	lock  *sync.RWMutex
}

// inMemoryQueueItem 请求队列项目
type inMemoryQueueItem struct {
	// Request 序列化的Request
	Request []byte

	// Next 指向队列中下个Request的指针
	Next *inMemoryQueueItem
}

// Init 初始化
func (q *InMemoryQueueStorage) Init() error {
	q.lock = &sync.RWMutex{}
	return nil
}

// AddRequest 向队列中添加序列化的Request
func (q *InMemoryQueueStorage) AddRequest(r []byte) error {
	q.lock.Lock()
	defer q.lock.Unlock()
	if q.MaxSize > 0 && q.size >= q.MaxSize { //超过了队列容量
		return colite.ErrQueueFull
	}
	i := &inMemoryQueueItem{Request: r}
	if q.first == nil {
		q.first = i
	} else {
		q.last.Next = i
	}
	q.last = i
	q.size++
	return nil
}

// GetRequest 从队列中获取序列化的Request
func (q *InMemoryQueueStorage) GetRequest() ([]byte, error) {
	q.lock.Lock()
	defer q.lock.Unlock()
	if q.size == 0 {
		return nil, nil
	}
	r := q.first.Request
	q.first = q.first.Next
	q.size--
	return r, nil
}

// QueueSize 队列大小
func (q *InMemoryQueueStorage) QueueSize() (int, error) {
	q.lock.Lock()
	defer q.lock.Unlock()
	return q.size, nil
}
